#ifndef DATAFLOW_EXECUTOR_H
#define DATAFLOW_EXECUTOR_H

#include <QObject>
#include <QThread>

#include "Graph.h"
#include "Scheduler.h"
#include "Worker.h"

namespace DataFlow
{

class Executor : public QObject
{
	Q_OBJECT

	public:

		Executor(const Graph & g, int workersCount = -1, QObject * parent = 0)
			: QObject(parent)
			, schedulerThread(0)
			, scheduler(0)
		{
			this->schedulerThread = new QThread();
			this->scheduler       = new Detail::Scheduler(g);

			this->connect
			(
				this->scheduler,
				SIGNAL(taskStarted(int,DataFlow::ArgumentArray)),
				SIGNAL(taskStarted(int,DataFlow::ArgumentArray))
			);

			this->connect
			(
				this->scheduler,
				SIGNAL(taskFinished(int,DataFlow::ArgumentArray,DataFlow::ArgumentArray)),
				SIGNAL(taskFinished(int,DataFlow::ArgumentArray,DataFlow::ArgumentArray))
			);

			this->connect
			(
				this->scheduler,
				SIGNAL(finished()),
				SIGNAL(finished())
			);

			this->scheduler->moveToThread(this->schedulerThread);
			this->schedulerThread->start();

			this->setWorkersCount(workersCount);
		}

		virtual ~Executor(void)
		{
			//this->cancelAndWait();

			this->setWorkersCount(0);

			this->schedulerThread->quit();
			this->schedulerThread->wait();

			delete this->scheduler;
			this->scheduler = 0;

			delete this->schedulerThread;
			this->schedulerThread = 0;
		}

		void setWorkersCount(int n)
		{
			if (n < 0) n = QThread::idealThreadCount();

			QMutexLocker locker(&(this->mutex));

			const int s = int(this->workers.size());
			if (n < s)
			{
				this->scheduler->dispatchTasks(false);
				for (int i=n; i<s; ++i)
				{
					delete this->workers[i];
				}
				this->scheduler->dispatchTasks(true);
				this->workers.resize(size_t(n));
			}
			else if (n > s)
			{
				this->workers.resize(size_t(n));
				for (int i=s; i<n; ++i)
				{
					this->workers[i] = new Detail::Worker(this->scheduler);
					this->workers[i]->start();
				}
			}
		}

		int workersCount(void) const
		{
			Executor * that = const_cast<Executor *>(this);
			QMutexLocker locker(&(that->mutex));
			return int(that->workers.size());
		}

		bool push(int task, int port, int id, const Argument & value)
		{
			this->scheduler->pushInput(task, port, id, value);
			return true;
		}

		void waitForFinished(void)
		{
			this->scheduler->waitForFinished();
		}

		/*
		void cancel(void)
		{
			;
		}

		void waitForFinished(void)
		{
			;
		}

		virtual void cancelAndWait(void)
		{
			this->cancel();
			this->waitForFinished();
		}
		*/

	signals:

		void taskStarted  (int task, DataFlow::ArgumentArray input);
		void taskFinished (int task, DataFlow::ArgumentArray input, DataFlow::ArgumentArray output);
		void finished     (void);

		/*
		void cancelled    (void);
		*/

	protected:

		QThread *                     schedulerThread;
		Detail::Scheduler *           scheduler;
		std::vector<Detail::Worker *> workers;
		QMutex                        mutex;
};

class VALUE
{
	public:

		VALUE(int id, const Argument & value)
			: i(id)
			, v(value)
		{
			;
		}

		int              id    (void) const { return this->i; }
		const Argument & value (void) const { return this->v; }

	protected:

		int      i;
		Argument v;
};

class DATA
{
	public:

		DATA(const INPUT & inputPin, const VALUE & value)
			: n(inputPin.node())
			, p(inputPin.port())
			, i(value.id())
			, v(value.value())
		{
			;
		}

		int              node  (void) const { return this->n; }
		int              port  (void) const { return this->p; }
		int              id    (void) const { return this->i; }
		const Argument & value (void) const { return this->v; }

	protected:

		int      n;
		int      p;
		int      i;
		Argument v;
};

inline Executor & operator << (Executor & executor, const DATA & data)
{
	(void)(executor.push(data.node(), data.port(), data.id(), data.value()));
	return executor;
}

}

#endif // DATAFLOW_EXECUTOR_H
